package com.zhang.gmall.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.zhang.gmall.utils.KafkaUtil;
import com.zhang.gmall.utils.MyEnv;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @title: 日志分流
 * @author: zhang
 * @date: 2022/3/11 18:28
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = MyEnv.getStreamingEnv(4, false);
        //TODO 3.读取kafka ods层数据，转换结构、过滤脏数据
        String topic = "ods_base_log_2022";
        String groupId = "base_log_app_dev";
        OutputTag<String> dirtyTag = new OutputTag<String>("Dirty-data") {
        };
        SingleOutputStreamOperator<JSONObject> kafkaDS = env
                .addSource(KafkaUtil.getKafkaSource(topic, groupId))
                .process(new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        try {
                            JSONObject jsonObject = JSON.parseObject(value);
                            out.collect(jsonObject);
                        } catch (Exception e) {
                            ctx.output(dirtyTag, value);
                        }
                    }
                });

        //TODO 4.新老用户修正
        SingleOutputStreamOperator<JSONObject> repairDS = kafkaDS
                .keyBy(data -> data.getJSONObject("common").getString("mid"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    //定义保存用户上次登录时间状态
                    private ValueState<String> timeState;
                    private SimpleDateFormat sdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        timeState = getRuntimeContext().getState(
                                new ValueStateDescriptor<String>(
                                        "time-state", Types.STRING
                                )
                        );

                        sdf = new SimpleDateFormat("yyyy-MM-dd");
                    }

                    @Override
                    public void processElement(JSONObject element, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        //获取用户is_new值来进行判断是否需要修复
                        String isNew = element.getJSONObject("common").getString("is_new");
                        //只有当不是新用户的时候才进行校验修复 1
                        if ("1".equals(isNew)) {
                            //获取状态上次登录状态
                            String lastTime = timeState.value();
                            Long ts = element.getLong("ts");
                            String curryTime = sdf.format(new Date(ts));
                            //TODO 小功能：如果业务需求只需对一天时间内无论登录多少次都是新访客，在if判断加上判断时期是否今天
                            //lastTime != null && lastTime.length() > 0 && !curryTime.equals(lastTime)
                            if (lastTime != null && lastTime.length() > 0 && !curryTime.equals(lastTime)) {
                                //修复
                                element.getJSONObject("common").put("is_new", 0);
                            } else {
                                timeState.update(curryTime);
                            }
                        }
                        out.collect(element);
                    }
                });

        //TODO 5.数据分流分5个流，页面日志--主流、曝光日志--侧输出流、启动日志--侧输出流、动作日志--侧输出流、错误日志--侧输出流
        //定义侧输出标签
        OutputTag<String> startTag = new OutputTag<String>("start") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("display") {
        };
        OutputTag<String> actionTag = new OutputTag<String>("action") {
        };
        OutputTag<String> errorTag = new OutputTag<String>("error") {
        };
        SingleOutputStreamOperator<String> resultDS = repairDS
                .process(new ProcessFunction<JSONObject, String>() {

                    @Override
                    public void processElement(JSONObject element, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                        String err = element.getString("err");
                        if (err != null) {
                            ctx.output(errorTag, element.toJSONString());
                        }
                        JSONObject startObj = element.getJSONObject("start");
                        if (startObj != null && startObj.size() > 0) {
                            //启动日志输出到侧输出流
                            ctx.output(startTag, element.toJSONString());

                        } else {
                            //不是启动日志即是页面日志，页面日志包括曝光日志
                            out.collect(element.toJSONString());
                            //提取pageId和ts
                            String pageId = element.getJSONObject("page").getString("page_id");
                            Long ts = element.getLong("ts");
                            JSONArray displayArray = element.getJSONArray("displays");
                            if (displayArray != null && displayArray.size() > 0) {
                                for (int i = 0; i < displayArray.size(); i++) {
                                    JSONObject displayObj = displayArray.getJSONObject(i);
                                    //为所有曝光信息添加页面id和ts字段
                                    displayObj.put("ts", ts);
                                    displayObj.put("page_id", pageId);
                                    ctx.output(displayTag, displayObj.toJSONString());
                                }
                            }

                            JSONArray actionArr = element.getJSONArray("actions");
                            if (actionArr != null && actionArr.size() > 0) {
                                for (int i = 0; i < actionArr.size(); i++) {
                                    JSONObject jsonObject = actionArr.getJSONObject(i);
                                    //为所有动作信息添加页面id和ts字段
                                    jsonObject.put("page_id", pageId);
                                    jsonObject.put("common", element.getJSONObject("common"));
                                    ctx.output(actionTag, jsonObject.toJSONString());
                                }
                            }

                        }
                    }
                });

        //TODO 6.提取侧输出流数据
        DataStream<String> startDS = resultDS.getSideOutput(startTag);
        DataStream<String> displayDS = resultDS.getSideOutput(displayTag);
        DataStream<String> errorDS = resultDS.getSideOutput(errorTag);
        DataStream<String> actionDS = resultDS.getSideOutput(actionTag);
        resultDS.print("Page>>>>>>>>");
        startDS.print("Start>>>>>>>>");
        actionDS.print("Action>>>>>>>>");
        displayDS.print("Display>>>>>>>>");
        errorDS.print("Error>>>>>>>>");
        //TODO 7.发送到kafka dwd层对应主题
        resultDS.addSink(KafkaUtil.getKafkaSink("dwd_page_log"));
        startDS.addSink(KafkaUtil.getKafkaSink("dwd_start_log"));
        errorDS.addSink(KafkaUtil.getKafkaSink("dwd_error_log"));
        displayDS.addSink(KafkaUtil.getKafkaSink("dwd_display_log"));
        actionDS.addSink(KafkaUtil.getKafkaSink("dwd_action_log"));
        //TODO 8.启动任务
        env.execute("BaseLogApp");
    }
}
